package com.yyx.demo.election;

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;

public class LeaderElectionSupport implements Watcher {
    private static final Logger LOG = LoggerFactory.getLogger(LeaderElectionSupport.class);

    private ZooKeeper zooKeeper;

    private State state;
    private Set<LeaderElectionAware> listeners;

    private String rootNodeName;
    private LeaderOffer leaderOffer;
    private String hostName;

    public synchronized void start(){
        state = State.START;

        //发布事件，供调用
        dispatchEvent(EventType.START);
        LOG.info("Starting leader election support");
        if (zooKeeper==null){
            throw new IllegalStateException("No instance of zookeeper provided.Hint : use setZooKeeper()");
        }
        if (hostName==null){
            throw new IllegalStateException(
                    "No hostname provided.Hint : use setHostName()");
        }

        try {
            makeOffer();
            determineElectionStatus();
        } catch (KeeperException | InterruptedException e) {
            becomeFailed(e);
        }
    }

    public synchronized void stop() {
        state = State.STOP;
        dispatchEvent(EventType.STOP_START);

        LOG.info("Stopping leader election support");

        if (leaderOffer != null) {
            try {
                zooKeeper.delete(leaderOffer.getNodePath(), -1);
                LOG.info("Removed leader offer {}", leaderOffer.getNodePath());
            } catch (InterruptedException | KeeperException e) {
                becomeFailed(e);
            }
        }

        dispatchEvent(EventType.STOP_COMPLETE);
    }

    private void becomeFailed(Exception e) {
        LOG.error("Failed in state {}", state, e);

        state = State.FAILED;
        dispatchEvent(EventType.FAILED);
    }

    private void makeOffer() throws KeeperException, InterruptedException {
        state =State.OFFER;
        dispatchEvent(EventType.OFFER_START);

        LeaderOffer newLeaderOffer = new LeaderOffer();
        byte[] hostnameBytes;
        synchronized (this){
            newLeaderOffer.setHostName(hostName);
            hostnameBytes = hostName.getBytes();
            newLeaderOffer.setNodePath(
                    zooKeeper.create(rootNodeName + "/" + "n_",
                            hostnameBytes, ZooDefs.Ids.OPEN_ACL_UNSAFE,
                            CreateMode.EPHEMERAL_SEQUENTIAL));
            leaderOffer=newLeaderOffer;
        }
        LOG.debug("Created leader offer {}", leaderOffer);

        dispatchEvent(EventType.OFFER_COMPLETE);
    }

    private void determineElectionStatus() throws KeeperException, InterruptedException {
        state = State.DETERMINE;
        dispatchEvent(EventType.DETERMINE_START);

        LeaderOffer currentLeaderOffer = getLeaderOffer();
        String[] components = currentLeaderOffer.getNodePath().split("/");

        currentLeaderOffer.setId(Integer.valueOf(components[components.length - 1].substring("n_".length())));

        List<LeaderOffer> leaderOffers = toLeaderOffers(zooKeeper.getChildren(rootNodeName, false));

        /*
         * For each leader offer, find out where we fit in. If we're first, we
         * become the leader. If we're not elected the leader, attempt to stat the
         * offer just less than us. If they exist, watch for their failure, but if
         * they don't, become the leader.
         */
        for (int i = 0; i < leaderOffers.size(); i++) {
            LeaderOffer leaderOffer = leaderOffers.get(i);

            if (leaderOffer.getId().equals(currentLeaderOffer.getId())) {
                LOG.debug("There are {} leader offers. I am {} in line.", leaderOffers.size(), i);

                dispatchEvent(EventType.DETERMINE_COMPLETE);

                if (i == 0) {
                    becomeLeader();
                } else {
                    becomeReady(leaderOffers.get(i - 1));
                }

                /* Once we've figured out where we are, we're done. */
                break;
            }
        }
    }

    private void becomeReady(LeaderOffer neighborLeaderOffer) throws KeeperException, InterruptedException {
        LOG.info(
                "{} not elected leader. Watching node: {}",
                getLeaderOffer().getNodePath(),
                neighborLeaderOffer.getNodePath());
        Stat stat = zooKeeper.exists(neighborLeaderOffer.getNodePath(),this);
        if (stat!=null){
            dispatchEvent(EventType.READY_START);
            state = State.READY;
            dispatchEvent(EventType.READY_COMPLETE);
        } else {
            LOG.info(
                    "We were behind {} but it looks like they died. Back to determination.",
                    neighborLeaderOffer.getNodePath());
            determineElectionStatus();
        }
    }

    private void becomeLeader() {
        state = State.ELECTED;
        dispatchEvent(EventType.ELECTED_START);

        LOG.info("Becoming leader with node: {}", getLeaderOffer().getNodePath());

        dispatchEvent(EventType.ELECTED_COMPLETE);
    }

    //进行排序
    private List<LeaderOffer> toLeaderOffers(List<String> strings)
            throws KeeperException, InterruptedException {

        List<LeaderOffer> leaderOffers = new ArrayList<>(strings.size());

        /*
         * Turn each child of rootNodeName into a leader offer. This is a tuple of
         * the sequence number and the node name.
         */
        for (String offer : strings) {
            String hostName = new String(zooKeeper.getData(rootNodeName + "/" + offer, false, null));

            leaderOffers.add(new LeaderOffer(
                    Integer.valueOf(offer.substring("n_".length())),
                    rootNodeName + "/" + offer, hostName));
        }

        /*
         * We sort leader offers by sequence number (which may not be zero-based or
         * contiguous) and keep their paths handy for setting watches.
         */
        Collections.sort(leaderOffers, new LeaderOffer.IdComparator());

        return leaderOffers;
    }

    private void dispatchEvent(EventType eventType) {
        LOG.debug("Dispatching event:{}",eventType);
        synchronized (listeners){
            if (listeners.size()>0){
                for (LeaderElectionAware observer: listeners){
                    observer.onElectionEvent(eventType);
                }
            }
        }
    }

    public LeaderElectionSupport() {
        state = State.STOP;
        //确保该集合的线程安全
        listeners = Collections.synchronizedSet(new HashSet<>());
    }

    private synchronized LeaderOffer getLeaderOffer() {
        return leaderOffer;
    }

    public String getRootNodeName() {
        return rootNodeName;
    }

    public void setRootNodeName(String rootNodeName) {
        this.rootNodeName = rootNodeName;
    }


    public ZooKeeper getZooKeeper() {
        return zooKeeper;
    }

    public void setZooKeeper(ZooKeeper zooKeeper) {
        this.zooKeeper = zooKeeper;
    }


    public synchronized String getHostName() {
        return hostName;
    }

    public synchronized void setHostName(String hostName) {
        this.hostName = hostName;
    }

    @Override
    public void process(WatchedEvent event) {
        if (event.getType().equals(Watcher.Event.EventType.NodeDeleted)) {
            if (!event.getPath().equals(getLeaderOffer().getNodePath())
                    && state != State.STOP) {
                LOG.debug(
                        "Node {} deleted. Need to run through the election process.",
                        event.getPath());
                try {
                    determineElectionStatus();
                } catch (KeeperException | InterruptedException e) {
                    becomeFailed(e);
                }
            }
        }
    }

    /**
     * The type of event.
     */
    public enum EventType {
        START,
        OFFER_START,
        OFFER_COMPLETE,
        DETERMINE_START,
        DETERMINE_COMPLETE,
        ELECTED_START,
        ELECTED_COMPLETE,
        READY_START,
        READY_COMPLETE,
        FAILED,
        STOP_START,
        STOP_COMPLETE,
    }

    /**
     * The internal state of the election support service.
     */
    public enum State {
        START,
        OFFER,
        DETERMINE,
        ELECTED,
        READY,
        FAILED,
        STOP
    }



}
